package com.wenx.kafkacamel;

import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.kafka.KafkaComponent;
import org.apache.camel.component.kafka.KafkaConstants;
import org.apache.camel.component.properties.PropertiesComponent;
import org.apache.camel.impl.DefaultCamelContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * camel get message from kafka
 */
public final class MessageConsumerClient2 extends RouteBuilder {

    private static final Logger LOG = LoggerFactory.getLogger(MessageConsumerClient2.class);

    private MessageConsumerClient2() {
    }

    public void configure() throws Exception {
        /*PropertiesComponent pc = getContext().getComponent("properties", PropertiesComponent.class);
        pc.setLocation("classpath:application-kafka.properties");*/

        log.info("About to start route: Kafka Server -> Log ");

        /*from("direct:start").process(new Processor() {

                        public void process(Exchange exchange) throws Exception {
                        exchange.getIn().setBody("Test Message from Camel Kafka Component Final",String.class);
                        exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, 0);
                        exchange.getIn().setHeader(KafkaConstants.KEY, "1");
                        System.out.println(exchange.getOut().getBody());
        }
        }).to("kafka:10.10.5.212:9092?topic=test1");*/

        from("kafka:10.10.5.212:9092?topic=test&groupId=test-consumer-group")
                .process(new Processor() {
                    public void process(Exchange exchange) throws Exception {
                        String aa = exchange.getIn().getBody().toString();
                        System.out.println(aa);
                    }
                });
                //.to("kafka:10.10.5.212:9092?topic=test1");
                //.log("${headers}");
    }

    public static void main(String[] args) throws Exception {

        CamelContext camelContext = new DefaultCamelContext();

        KafkaComponent kafka = new KafkaComponent();
        kafka.setBrokers("10.10.5.212:9092");

        camelContext.addComponent("kafka", kafka);

        camelContext.addRoutes(new MessageConsumerClient2());

        camelContext.start();

        // let it run for 5 minutes before shutting down
        Thread.sleep(5 * 60 * 1000);

        camelContext.stop();
    }

}